import model.Utils
import model.configs.{KafkaConfig, StorageWorkerConfig, VkConfig}
import model.events.BaseEvent
import zio.json.*
import zio.kafka.*
import zio.kafka.consumer.{Consumer, ConsumerSettings, OffsetBatch, Subscription}
import zio.kafka.serde.Serde
import zio.Console.*
import zio.{ZIO, ZIOAppDefault, ZLayer}
import model.vk.UserInfo
import model.vk.UserInfo.UserInfoWrapper
import zio.*
import zio.config.magnolia.deriveConfig
import zio.config.typesafe.TypesafeConfigProvider
import zio.config.typesafe.FromConfigSourceTypesafe

object StorageWorker extends ZIOAppDefault {
  private val consumer =
    ZLayer.fromZIO(
      for {
        kafkaConfig <- TypesafeConfigProvider.fromResourcePath().load(deriveConfig[KafkaConfig])
        consumer <- Consumer.make(
          ConsumerSettings(kafkaConfig.bootstrapServers).withGroupId(
            kafkaConfig.group.getOrElse("storage-worker")
          )
        )
      } yield consumer
    )

  override val bootstrap: ZLayer[ZIOAppArgs, Any, Any] =
    Runtime.setConfigProvider(ConfigProvider.fromResourcePath())

  private def matchPersist(
      topic: String
  ): List[BaseEvent] => ZIO[DBService.Service, Throwable, List[Long]] = topic match {
    case "like_add"       => DBService.persistLikesAdd
    case "like_remove"    => DBService.persistLikesRemove
    case "group_leave"    => DBService.persistGroupLeaves
    case "group_join"     => DBService.persistGroupJoins
    case "wall_repost"    => DBService.persistWallReposts
    case "wall_reply_new" => DBService.persistWallRepliesNew
  }

  private def buildConsumer(topic: String, batchSize: Int, groupedDuration: Duration) =
    Consumer
      .plainStream(Subscription.topics(topic), Serde.long, Serde.string)
      .groupedWithin(batchSize, groupedDuration)
      .tap(batch =>
        val (jsonErrors, events) = batch.toList
          .map(event => Utils.ConvertToJsonByTopic(topic, event.value))
          .partitionMap(identity)

        // TODO: отрицательные значения -- это значит, что действие совершил не пользователь, а группа/паблик
        val filteredEvents = events.filter(_.getUserId.get > 0)
        for {
          _ <- ZIO.logInfo(s"batch size = ${batch.size}, events = ${filteredEvents.mkString}")
          vkSettings <- TypesafeConfigProvider.fromResourcePath().load(deriveConfig[VkConfig])
          usersInfoString <- VkApiUtil.fetchUserInfo(
            filteredEvents.flatMap(_.getUserId).mkString(","),
            vkSettings
          )
          usersListWrapper <- ZIO.fromEither(usersInfoString.fromJson[UserInfoWrapper])
          _                <- DBService.persistUsersInfo(usersListWrapper.response)
          // если не получится положить в бд user info, то запишется ли информация об ивенте?
          _ <- matchPersist(topic)(filteredEvents)
          _ <- ZIO.logInfo(
            s"userInfoString = $usersInfoString, events user id = ${filteredEvents.map(_.getUserId).mkString(",")}"
          )
        } yield ()
      )
      .map { batch => batch.map(_.offset).foldLeft(OffsetBatch.empty)(_ add _) }
      .mapZIO(_.commit)
      .drain

  def run =
    for {
      config <- TypesafeConfigProvider.fromResourcePath().load(deriveConfig[StorageWorkerConfig])
      consumers <- config.consumingTopics
        .map(topic => buildConsumer(topic, config.batchSize, config.groupedDuration))
        .reduce(_ merge _)
        .runDrain
        .provideLayer(DBService.live ++ consumer)
    } yield consumers
}
